package Flink.split;

import Flink.bean.WaterSensor;
import Flink.function.WaterSensorMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class SplitStreamByOutputTag {
    // Todo 将WaterSensor按照Id类型进行分流
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<WaterSensor> ds = env
                .socketTextStream("master", 7777)
                .map(new WaterSensorMapFunction());
        OutputTag<WaterSensor> s1 = new
                OutputTag<WaterSensor>("s1", Types.POJO(WaterSensor.class)) {};
        OutputTag<WaterSensor> s2 = new
                OutputTag<WaterSensor>("s2", Types.POJO(WaterSensor.class)) {};

        SingleOutputStreamOperator<WaterSensor> ds1 = ds
                .process(new ProcessFunction<WaterSensor, WaterSensor>() {
                    @Override
                    public void processElement(WaterSensor value, ProcessFunction<WaterSensor, WaterSensor>.Context ctx, Collector<WaterSensor> out) throws Exception {
                        if ("s1".equals(value.getId())) {
                            ctx.output(s1, value);
                        } else if ("s2".equals(value.getId())) {
                            ctx.output(s2, value);
                        } else {
                            out.collect(value);
                        }
                    }
                });
        // 每次输出前都会加上”主流，非s1,s2的传感器“这句话
        ds1.print("主流，非s1,s2的传感器");
        // 从主流中获取测流
        SideOutputDataStream<WaterSensor> sideOutput1 = ds1.getSideOutput(s1);
        SideOutputDataStream<WaterSensor> sideOutput2 = ds1.getSideOutput(s2);

        sideOutput1.printToErr("s1");
        sideOutput2.printToErr("s2");

        env.execute();
    }
}
